package com.citusdata.elven.linearRoad;

import com.citusdata.elven.datastore.protobuf.Datastore.*;
import com.citusdata.elven.datastore.BdbDatastore;
import com.citusdata.elven.linearRoad.protobuf.LinearRoad.*;
import com.citusdata.elven.linearRoad.ConversionUtil;
import com.citusdata.elven.linearRoad.FullPosition;
import com.citusdata.elven.linearRoad.SegmentStatistics;
import com.citusdata.elven.linearRoad.VehicleInformation;
import com.citusdata.elven.util.Stopwatch;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;


// Helper class to convert seconds into minutes; the conversion follows the
// Linear Road benchmark specification.
class Time {
  public static int getMinute(int second) {
    // In an inexact integer division, Java rounds the result down.
    return ((second/60) + 1);
  }
}


// Service class that implements the required functionality for the Linear Road
// benchmark. At a high level, this class processes two types of queries: 
//
// (1) Realtime queries that are "computation heavy", namely position report
// queries. All the state required to process these queries are stored in this
// class. As a result, distributed implementations are responsible for
// forwarding all position reports that belong to a certain "partition" to the
// same LinearRoadService object. In a typical Linear Road benchmark run, 99% of
// all input queries are of this type.
//
// (2) Historical queries that are "storage heavy", namely account balance and
// daily expenditure queries. Note that along with implementations to process
// historical queries, this class provides wrapper methods to read from and
// write to a local Berkeley DB store. These methods are just implemented to
// provide a functional non-httpRpc (local) implementation, and are _not_ used
// by distributed implementations.


public class LinearRoadService extends LinearRoadServiceInterface {
  private static final int    ACCIDENT_DOWNSTREAM = 4;
  private static final int    START_MINUTE = 0;
  private static final int    TOLLED_AVERAGE_MINS = 5;
  private static final double TOLLED_SPEED  = 40.0;
  private static final int    TOLLED_VEHICLES = 50;

  private SegmentStatistics segmentStats   = new SegmentStatistics();
  private AccidentDetection accidentDetect = new AccidentDetection();
  private ConcurrentMap vehicleMap  = new ConcurrentHashMap();

  private List eventCounts;
  private final BdbDatastore localDatastore;

  public LinearRoadService() {
    this.localDatastore = null;
    initEventCounts();
  }

  public LinearRoadService(BdbDatastore localDatastore) {
    this.localDatastore = localDatastore;
    initEventCounts();
  }

  // ** The following methods implement computation-related functionality for
  // LinearRoad. More specifically, these methods are response for handling
  // Position Report queries.

  protected static boolean exitLane(int lane) {
    return (lane == 4);
  }


  private void initEventCounts() {
    eventCounts = new ArrayList();

    for (int min = 0; min < SegmentStatistics.TOTAL_MINS; ++min) {
      AtomicInteger zeroCount = new AtomicInteger();
      eventCounts.add(zeroCount);
    }
  }


  private AccidentNotification updateAndDetectAccidents(PositionRequest req) {
    Stopwatch stopwatch = new Stopwatch();
    stopwatch.start();

    // Use the "get then putIfAbsent" idiom to access/update each vehicle.
    int vehicleId = req.getVid();
    VehicleInformation vehicle = (VehicleInformation) vehicleMap.get(vehicleId);
    if (vehicle == null) {
      VehicleInformation newVehicle = new VehicleInformation();
      vehicle = (VehicleInformation) vehicleMap.putIfAbsent(vehicleId, newVehicle); 
      if (vehicle == null) {
        vehicle = newVehicle;
      }
    }

    int curTime = req.getTime();
    int curMinute = Time.getMinute(curTime);
    FullPosition position = new FullPosition(req.getXway(), req.getDir(), 
                                             req.getPos()); 

    // This function should be called _first_ on the vehicle object.
    vehicle.recordLocation(curTime, position, req.getLane());
    if (vehicle.justStopped(curTime)) {
      accidentDetect.recordStoppedVehicle(curMinute, vehicleId, position); 
    } else if (vehicle.justMoved(curTime)) {
      accidentDetect.clearStoppedVehicle(curMinute, vehicleId);
    } else {
      ;  // Nothing important happened.
    }

    FullSegment segment = new FullSegment(req.getXway(), req.getDir(), 
                                          req.getSeg());

    for (int count = 0; count <= ACCIDENT_DOWNSTREAM; ++count) {
      FullSegment fullDownstream = FullSegment.downstreamSegment(segment, count);
      int downstream = fullDownstream.getSegment();
      int begin      = ConversionUtil.segBeginToPos(downstream);
      int end        = ConversionUtil.segEndToPos(downstream);

      FullPosition posBegin = new FullPosition(req.getXway(), req.getDir(), begin);
      FullPosition posEnd   = new FullPosition(req.getXway(), req.getDir(), end);

      if (accidentDetect.accidentExists(curMinute, posBegin, posEnd)) {
        AccidentNotification.Builder accident = AccidentNotification.newBuilder(); 
        accident.setTime(req.getTime());
        accident.setEmitTime(req.getTime() + (int) stopwatch.elapsedSeconds());
        accident.setVid(req.getVid());
        accident.setAccidSegment(downstream);

        return accident.build();
      }
    }

    // Otherwise, there were no accidents
    return null;
  }

  
  private TollNotification updateAndComputeToll(PositionRequest req) {
    Stopwatch stopwatch = new Stopwatch();
    stopwatch.start();

    FullSegment segment = new FullSegment(req.getXway(), req.getDir(), req.getSeg());
    int curMinute = Time.getMinute(req.getTime());

    segmentStats.addVehicleSpeed(curMinute, segment, req.getVid(), req.getSpeed());

    // Make sure that we do not try to access information that occurred before
    // the simulation started.
    int prevMinute = Math.max(curMinute-1, START_MINUTE);
    int vehicleCount = segmentStats.vehicleCount(prevMinute, segment);
    double speedSum = 0.0;

    int time = Math.max(curMinute-TOLLED_AVERAGE_MINS, START_MINUTE);
    int segmentCount = 0;
    for (; time <= prevMinute; ++time) {
      if (segmentStats.vehicleCount(time, segment) > 0) {
        segmentCount++;
        speedSum += segmentStats.speedAverage(time, segment);
      }
    }

    double speedAverage = 0.0;
    if (segmentCount == 0) {
       // Expected default value if no previous <segment,times> were initialized
      speedAverage = -1.0; 
    } else {
      speedAverage = (speedSum / segmentCount);
    }

    int segmentToll = 0;
    if (vehicleCount > TOLLED_VEHICLES && speedAverage < TOLLED_SPEED) {
      int tolledVehicleCount = vehicleCount-TOLLED_VEHICLES;
      segmentToll = 2*tolledVehicleCount*tolledVehicleCount;
    }

    TollNotification.Builder toll = TollNotification.newBuilder();
    toll.setVid(req.getVid());
    toll.setTime(req.getTime());
    toll.setEmitTime(req.getTime() + (int) stopwatch.elapsedSeconds());
    toll.setSegSpeed((int) speedAverage);
    toll.setToll(segmentToll);

    return toll.build();
  }


  private TollAssessment constructAssessment(PositionRequest req, int previousToll) {
    TollAssessment.Builder assessment = TollAssessment.newBuilder();
    assessment.setVid(req.getVid());
    assessment.setTime(req.getTime());
    assessment.setXway(req.getXway());
    assessment.setToll(previousToll);

    return (assessment.build());
  }
 

  private int pastEventCountSum(int curMinute, int pastMinuteCount) {
    // Make sure that we do not try to access information that occurred before
    // the simulation started.
    int startMinute = Math.max(curMinute - pastMinuteCount, START_MINUTE);
    int endMinute   = Math.max(curMinute - 1, START_MINUTE);

    int totalCount = 0;
    for (int min = startMinute; min <= endMinute; ++min) {
      AtomicInteger eventCount = (AtomicInteger) eventCounts.get(min);
      totalCount += eventCount.get();
    }

    return totalCount;
  }


  public PositionResponse reportPosition(PositionRequest req) {
    TollNotification toll = updateAndComputeToll(req);
    AccidentNotification accident = updateAndDetectAccidents(req);
    PositionResponse.Builder response = PositionResponse.newBuilder();

    // VehicleInformation should already be recorded in previous calls.
    VehicleInformation vehicle = (VehicleInformation) vehicleMap.get(req.getVid()); 

    if (vehicle.changedSegments(req.getTime()) && !exitLane(req.getLane())) {
      int futureToll;

      if (accident != null) {
        futureToll = 0;

        // If there was an accident, _override_ toll and set it to 0.
        TollNotification.Builder newToll = TollNotification.newBuilder();
        newToll.mergeFrom(toll);
        newToll.setToll(0);
        
        response.setToll(newToll.build());
        response.setAccident(accident);
      } else {
        futureToll = toll.getToll();
        response.setToll(toll);
      }

      int previousToll = vehicle.getTollAssessment(req.getTime());
      TollAssessment debitToll = constructAssessment(req, previousToll);
      response.setAssessment(debitToll);

      vehicle.setTollAssessment(req.getTime(), futureToll);
    }

    int curMinute = Time.getMinute(req.getTime());
    AtomicInteger eventCount = (AtomicInteger) eventCounts.get(curMinute);
    eventCount.incrementAndGet();
    
    int totalCount = pastEventCountSum(curMinute, TOLLED_AVERAGE_MINS);
    response.setEventCount(totalCount);

    return response.build();
  }


  // Interface method for remote service calls
  public void reportPosition(RpcController controller, PositionRequest request,
                             RpcCallback done) {
    PositionResponse response = reportPosition(request);
    done.run(response);
  }

  
  // ** The following methods implement storage-related functionality for
  // LinearRoad. More specifically, these methods are response for handling
  // Account Balance and Daily Expenditure queries. 

  public boolean localStorage() {
    boolean localStorage = (localDatastore != null);
    return localStorage;
  }


  public void putHistoricalToll(int vehicleId, int time, TollAssessment histToll) {
    String vehicleKey = Integer.toString(vehicleId);

    PutRequest.Builder putBuilder = PutRequest.newBuilder();
    putBuilder.setPartitionKey(vehicleKey);
    putBuilder.setTimestamp((long) time);
    putBuilder.setData(histToll.toByteString());
    PutRequest putRequest = putBuilder.build();

    localDatastore.put(putRequest);
  }


  public AccountResponse computeAccount(AccountRequest accReq) {
    int epochStart = 0;
    List tolls = getHistoricalTolls(accReq.getVid(), epochStart, accReq.getTime());

    AccountResponse response = computeAccount(accReq, tolls);
    return response;
  }


  public ExpenditureResponse computeExpenditure(ExpenditureRequest expReq) {
    // Time for historical queries are represented as "negative days"
    int adjustedDay = expReq.getDay() * -1;
    List tolls = getHistoricalTolls(expReq.getVid(), adjustedDay, adjustedDay);

    ExpenditureResponse response = computeExpenditure(expReq, tolls);
    return response;
  }

  
  protected static AccountResponse computeAccount(AccountRequest accReq, 
                                                  List historicalTolls) {
    Stopwatch stopwatch = new Stopwatch();
    stopwatch.start();

    int totalBalance = 0;
    int lastUpdateTime = 0;

    for (int i = 0; i < historicalTolls.size(); ++i) {
      TollAssessment histToll = (TollAssessment) historicalTolls.get(i);
      
      lastUpdateTime = histToll.getTime();
      totalBalance += histToll.getToll();
    }

    AccountResponse.Builder response = AccountResponse.newBuilder();
    response.setTime(accReq.getTime());
    response.setEmitTime(accReq.getTime() + (int) stopwatch.elapsedSeconds());
    response.setQid(accReq.getQid());
    response.setResultTime(lastUpdateTime);
    response.setBalance(totalBalance);
    return response.build();
  }

  
  protected static ExpenditureResponse computeExpenditure(ExpenditureRequest expReq, 
                                                          List historicalTolls) {
    Stopwatch stopwatch = new Stopwatch();
    stopwatch.start();

    int dailyBalance = 0;
    if (historicalTolls.size() > 0) {
      // There should only be one assessment per day
      TollAssessment dailyToll = (TollAssessment) historicalTolls.get(0);

      int queriedExpressway = expReq.getXway();
      if (queriedExpressway == dailyToll.getXway()) {
        dailyBalance = dailyToll.getToll();
      } else {
        return null;
      }
    } else {
      return null;
    }

    ExpenditureResponse.Builder response = ExpenditureResponse.newBuilder();
    response.setTime(expReq.getTime());
    response.setEmitTime(expReq.getTime() + (int) stopwatch.elapsedSeconds());
    response.setQid(expReq.getQid());
    response.setBalance(dailyBalance);
    return response.build();
  }


  protected static List parseHistoricalTolls(List putRequests) {
    List historicalTolls = new ArrayList();

    try {
      for (int i = 0; i < putRequests.size(); ++i) {
        PutRequest putRequest = (PutRequest) putRequests.get(i);
        TollAssessment histToll = TollAssessment.parseFrom(putRequest.getData());

        historicalTolls.add(histToll);
      }
    } catch (InvalidProtocolBufferException ex) {
      System.err.println("Could not parse message, some tolls are missing");
    }

    return historicalTolls;
  }


  private List getHistoricalTolls(int vehicleId, int startTime, int endTime) {
    String vehicleKey = Integer.toString(vehicleId);

    GetRangeRequest.Builder getBuilder = GetRangeRequest.newBuilder();
    getBuilder.setPartitionKey(vehicleKey);
    getBuilder.setStartTime(startTime);
    getBuilder.setEndTime(endTime);
    GetRangeRequest getRequest = getBuilder.build();

    GetRangeResponse getResponse = localDatastore.getRange(getRequest);
    List putRequests = getResponse.getPutRequestsList();

    List historicalTolls = parseHistoricalTolls(putRequests);
    return historicalTolls;
  }
}
